package com.apps.sdses.flink141.cdc;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/6/13 23:04
 */
public class MySQLCDC2Print {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

//        tableEnv.executeSql("CREATE TABLE products (\n" +
//                "    id INT,\n" +
//                "    name STRING,\n" +
////                "    description STRING,\n" +
//                "    PRIMARY KEY (id) NOT ENFORCED\n" +
//                "  ) WITH (\n" +
//                "    'connector' = 'mysql-cdc',\n" +
//                "    'hostname' = 'localhost',\n" +
//                "    'port' = '3306',\n" +
//                "    'username' = 'root',\n" +
//                "    'password' = '123456',\n" +
////                "    'server-time-zone' = '+08:00',\n" +
//                "    'server-time-zone' = 'Asia/Shanghai',\n" +
//                "    'database-name' = 'mydb',\n" +
//                "    'table-name' = 'test_table'\n" +
////                "    'scan.startup.mode' = 'latest-offset',\n" +
//
////                "    'scan.startup.mode' = 'timestamp',\n" +
////                "    'scan.startup.timestamp-millis' = '1689256818000',\n" +
////                "    'server-id' = '8637102'\n" +
//                "  )");



        tableEnv.executeSql("CREATE TABLE water_sensor_test (\n" +
                "    id INT,\n" +
                "    mode STRING,\n" +
                "    voltage FLOAT,\n" +
                "    resistance FLOAT,\n" +
                "    temperature FLOAT,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                "  ) WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = '192.168.102.155',\n" +
                "    'port' = '33066',\n" +
                "    'username' = 'datacube_test',\n" +
                "    'password' = 'Gbd#XERA@19d1ZtD',\n" +
//                "    'server-time-zone' = '+08:00',\n" +
                "    'server-time-zone' = 'Asia/Shanghai',\n" +
                "    'database-name' = 'db_datacube_test',\n" +
                "    'table-name' = 'water_sensor_test'\n" +
//                "    'scan.startup.mode' = 'latest-offset',\n" +

//                "    'scan.startup.mode' = 'timestamp',\n" +
//                "    'scan.startup.timestamp-millis' = '1689256818000',\n" +
//                "    'server-id' = '8637102'\n" +
                "  )");




        tableEnv.executeSql("CREATE TABLE print_table WITH ('connector' = 'print')\n" +
                "LIKE water_sensor_test (EXCLUDING ALL)");


        tableEnv.executeSql("insert into print_table select * from water_sensor_test ");


    }
}
